package worker;

import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;

import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import main.Block;
import main.Partition;
import main.Simulate;

/**
 * 
 * Tries to split the splitBlock.
 * 
 */
public class SplitWorker implements Runnable {

	/**
	 * Block we want to split
	 */
	private Block splitBlock;
	/**
	 * partition
	 */
	private Partition partition;
	/**
	 * number of split blocks currently processed.
	 */
	private Semaphore numberResults;

	/**
	 * Creates a new instance of splitWorker.
	 * 
	 * @param splitBlock
	 *            block to split.
	 * @param partition
	 *            partition
	 * @param numberResults
	 *            number of splitBlocks currently processed.
	 */
	public SplitWorker(Block splitBlock, Partition partition,
			Semaphore numberResults) {

		this.splitBlock = splitBlock;
		this.partition = partition;
		this.numberResults = numberResults;

	}

	/**
	 * Tries to split the splitBlock.
	 */
	public void run() {

		// First get all post-blocks. Any outgoing edge from the splitBlock
		// points to some post-block. Only those blocks may split the
		// split-block.
		List<Block> postBlocks = partition.getPostBlocks(splitBlock);
		// the current splitBlock is not split.
		boolean splitted = false;

		// Create a new thread-pool. Try to find some post-block, that splits
		// split-block. We can find this post-block concurrently.
		ThreadPoolExecutor postWorkerPool = new ThreadPoolExecutor(
				Simulate.postWorkerPoolCore, Simulate.postWorkerPoolMax, 500,
				TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());

		// number of submitted tasks. Initially this number is 0. When a
		// post-block is submitted to thread-pool the number of tasks grows.
		// If we want to get the result of the task, then we count down this
		// counter.
		Semaphore taskCounter = new Semaphore(0);
		// Counts the number of returned results.
		Semaphore resultCounter = new Semaphore(0, true);
		// Counts the number of results, that ended up in split. PostWorker
		// ensures, that there is at most one positiveResult.
		Semaphore positiveResult = new Semaphore(0, true);

		// Submit all post-blocks.
		for (Block postBlock : postBlocks) {

			taskCounter.release();

			postWorkerPool.execute(new PostWorker(splitBlock, partition,
					postBlock, resultCounter, positiveResult));

		}

		try {
			// If task-counter is greater than 0, then we acquire one ticket,
			// else we are done, since there are no more task results.
			while (taskCounter.tryAcquire()) {

				// Here we wait for the result from one of the threads.
				resultCounter.acquire();
				// Now the result is obtained, but check if the split-block is
				// split. If it is split, then some thread returned positive
				// result.
				if (positiveResult.tryAcquire()) {

					// if it is the case, then we are done with this block.
					// Shutdown threadPool.
					postWorkerPool.shutdownNow();
					// Block is split.
					splitted = true;
					// leave while.
					break;
				}

			}

		} catch (InterruptedException ex) {
			// we where interrupted.
			ex.printStackTrace();
		}

		// if the split-block is NOT split.
		if (!splitted) {

			// then this block is stable at this time. And we release this
			// block, such that other threads later may have some new try if the
			// block gets unstable again.
			splitBlock.setStableAndRelease();
		}

		// one new split-block is now processed. So we have one new additional
		// result.
		this.numberResults.release();

	}
}
